package com.atguigu.flink.cep;

import com.atguigu.flink.function.MyUtil;
import com.atguigu.flink.pojo.LoginEvent;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.Map;

/**
 * Created by Smexy on 2022/12/17
 *
 * 如果同一用户（可以是不同IP）在2秒之内连续两次登录，都失败，就认为存在恶意登录的风险，输出相关的信息进行报警提示
 *
 *
 */
public class Demo13_CEPLogin
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        WatermarkStrategy<LoginEvent> watermarkStrategy = WatermarkStrategy.<LoginEvent>forMonotonousTimestamps()
            .withTimestampAssigner( (e, r) -> e.getEventTime() * 1000);

        //env.setParallelism(1);

        KeyedStream<LoginEvent, Long> ds = env
            .readTextFile("data/LoginLog.csv")
            .map(new MapFunction<String, LoginEvent>()
            {
                @Override
                public LoginEvent map(String value) throws Exception {
                    String[] words = value.split(",");
                    LoginEvent event = new LoginEvent(
                        Long.valueOf(words[0]),
                        words[1],
                        words[2],
                        Long.valueOf(words[3])
                    );
                    return event;
                }
            })
            .assignTimestampsAndWatermarks(watermarkStrategy)
            .keyBy(LoginEvent::getUserId);


        //定义规则
        Pattern<LoginEvent, LoginEvent> pattern = Pattern.<LoginEvent>begin("fail")
            .where(new SimpleCondition<LoginEvent>()
            {
                @Override
                public boolean filter(LoginEvent value) throws Exception {
                    return "fail".equals(value.getEventType());
                }
            })
            .times(2)
            .consecutive()  //紧挨着对两次都是失败
            //下一条不在2s内到达，属于超时。超时不取，取未超时，正常匹配
            .within(Time.seconds(2));

        CEP.pattern(ds,pattern)
           .select(new PatternSelectFunction<LoginEvent, String>()
           {
               /*
                    key:  规则名 fail
                    value:   List<LoginEvent>> pattern  匹配到的数据。
                                如果匹配成功一定是2条
                */
               @Override
               public String select(Map<String, List<LoginEvent>> map) throws Exception {
                   Long userId = map.get("fail").get(0).getUserId();
                   return userId + "恶意登录...";
               }
           }).print().setParallelism(1);

        env.execute();

    }
}
